package net.qa.demo.rocketmq;

import java.nio.charset.Charset;
import java.util.List;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class BroadcastExample {
	private static final Logger log = LoggerFactory.getLogger(BroadcastExample.class);
	
	static DefaultMQProducer producer;
	static DefaultMQPushConsumer consumer;
	static String currentMsgContent;
	
	@BeforeClass
	public static void setUpBeforeClass() throws Exception {
		
		//构造消息订阅者
		DefaultMQPushConsumer defConsumer = new DefaultMQPushConsumer(MQConfig.GROUP);
		defConsumer.setNamesrvAddr(MQConfig.NAME_SERVICE_ADDR);
		defConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		defConsumer.setMessageModel(MessageModel.BROADCASTING);
		defConsumer.subscribe(MQConfig.TOPIC, "*");
		defConsumer.registerMessageListener(new MessageListenerOrderly() {
			@Override
			public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
				
				log.info("{} Message Received[{}] : ", MQConfig.TOPIC, msgs.size());
				msgs.forEach(msg -> {
					String msgContent = new String(msg.getBody(), Charset.forName(RemotingHelper.DEFAULT_CHARSET));
					log.info("\t Message Handled[{}] : {}", msg.getMsgId(), msgContent);
					currentMsgContent = msgContent;
				});
				
				return ConsumeOrderlyStatus.SUCCESS;
			}
		});
		
		defConsumer.start();
		consumer = defConsumer;
		
		//构造消息生产者
        DefaultMQProducer defProducer = new DefaultMQProducer(MQConfig.GROUP);
        defProducer.setNamesrvAddr(MQConfig.NAME_SERVICE_ADDR);
        defProducer.setInstanceName(MQConfig.INSTANCE);
        defProducer.start();
                
        producer = defProducer;
        
	}

	@AfterClass
	public static void tearDownAfterClass() throws Exception {
		
		producer.shutdown();
		consumer.shutdown();
		
	}

	@Before
	public void setUp() throws Exception {
		
	}

	@After
	public void tearDown() throws Exception {
	}
	
	@Test //可靠的同步传输
	public void testReliableSync() throws Exception {
		
		String msgContent = "Reliable synchronous transmission"; 
        Message msg = new Message(MQConfig.TOPIC, MQConfig.TAG,
        		msgContent.getBytes(RemotingHelper.DEFAULT_CHARSET));
        SendResult sendResult = producer.send(msg);
        
        log.info("Reliable synchronous transmission: \n\t{}", sendResult);
        
        for (int i=0; i<10; i++) {
        	if (currentMsgContent != null) break;
        	Thread.sleep(1000);
        }
        Assert.assertEquals(currentMsgContent, msgContent);
        currentMsgContent = null;
        
	}
	
	@Test //可靠的异步传输
	public void testReliableAsync() throws Exception {
		
		String msgKey = "MyMessageKey-001";
		String msgContent = "Reliable asynchronous transmission";
		
        Message msg = new Message(MQConfig.TOPIC, MQConfig.TAG, msgKey, 
        		msgContent.getBytes(RemotingHelper.DEFAULT_CHARSET));
        
        producer.send(msg, new SendCallback() {
			
			@Override
			public void onSuccess(SendResult sr) {
				log.info("Reliable asynchronous transmission: \n\t{}", sr);
			}
			
			@Override
			public void onException(Throwable e) {
				log.warn("Reliable asynchronous transmission: fail", e);
			}
		});
        
        for (int i=0; i<10; i++) {
        	if (currentMsgContent != null) break;
        	Thread.sleep(1000);
        }
        Assert.assertEquals(currentMsgContent, msgContent);
        currentMsgContent = null;
        
	}
	
	@Test //中等可靠的单向传输
	public void testOneway() throws Exception {
		String msgContent = "Oneway transmission"; 
        Message msg = new Message(MQConfig.TOPIC, MQConfig.TAG, msgContent.getBytes(RemotingHelper.DEFAULT_CHARSET)); 
        producer.sendOneway(msg);
        
        log.info("Oneway transmission: \n\t{}", msg);
        
        for (int i=0; i<10; i++) {
        	if (currentMsgContent != null) break;
        	Thread.sleep(1000);
        }
        Assert.assertEquals(currentMsgContent, msgContent);
        currentMsgContent = null;
        
	}
	
	
}
